
import numpy as np
from math import sqrt, floor, ceil
import matplotlib.pyplot as plt
import dataDbControls as db
import time
import datasource.mairuiapi as mr
import txd.txd as tx
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
import logging

"""使用APScheduler配置Cron表达式任务"""
scheduler = BackgroundScheduler()


import utils.oss as oss
import pandas as pd
from io import StringIO
import utils.attribute_conversion as ac
def oss_soure_data(row):
    """
        acrossDays 是否跨天  否 则取startDay == day  否则就是范围
        开始日
        结束日
        股票代码
        时间维度

        timeds =timed.split("m")[0]
    """
    # acrossDays= row['across_days']
    # startDay= row['across_days']
    # endDay= row['across_days']
    stockCode= row['stock_code']
    td= row['time_period']

    x = oss.OSSDataProcessor();
    etf_code = stockCode.split(".")[0]
    key=stockCode+"/"+etf_code+"_"+td+".jsonl";
    select_json_params = {'Json_Type': 'LINES'}

    lastQuantity = row['last_quantity']

    # 类型  条件范围


    result = x.bucket.select_object(key, "select * from ossobject s limit "+str(lastQuantity), None, select_json_params)
    df = pd.read_json(StringIO(result.read().decode('utf-8')), lines=True)
    df =df.sort_values('timestamp')
    key_mapping = {'t': "时间", 'o': "开盘价", 'h': '最高价', 'l': '最低价', 'c': '收盘价', 'v': '成交量', 'a': "成交额"}

    df =ac.dataKeyConversion(df,key_mapping)
    return df

import threading

# 存储运行中的任务
running_tasks = {}
task_lock = threading.Lock()

def setup_logging():
    """设置日志"""
    logging.basicConfig(
        level=logging.INFO,
        format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    return logging.getLogger(__name__)

logger = setup_logging()



import uuid
from flask import Flask, jsonify, request
import datetime


def permanent_task():
    """
    永久运行的后台任务
    """
    task_id =1
    logger.info(f"任务 {task_id} 启动")
    try:

        init_task_zb()

        # 任务主循环
        # while running_tasks.get(task_id, {}).get('running', False):
        #     # 执行任务逻辑
        #     start_time = time.time()
        #
        #     # 模拟任务处理
        #     logger.info(f"任务 {task_id} 执行中...")
        #
        #     # 这里替换为您的实际任务逻辑
        #     # 例如：监控股票数据、处理数据流等
        #
        #     # 更新任务状态
        #     with task_lock:
        #         if task_id in running_tasks:
        #             running_tasks[task_id]['last_execution'] = datetime.now().isoformat()
        #             running_tasks[task_id]['execution_count'] += 1
        #
        #     # 模拟处理时间
        #     time.sleep(5)  # 可以根据需要调整
        #
        #     # 检查是否应该停止
        #     if not running_tasks.get(task_id, {}).get('running', False):
        #         break

    except Exception as e:
        logger.error(f"任务 {task_id} 执行错误: {e}")
    finally:
        # 清理任务
        with task_lock:
            if task_id in running_tasks:
                running_tasks[task_id]['running'] = False
                running_tasks[task_id]['end_time'] = datetime.now().isoformat()
        logger.info(f"任务 {task_id} 已停止")



def task_init():
    """启动永久任务"""
    try:
        # 生成唯一任务ID
        task_id =12335
        #str(uuid.uuid4())
        # 获取配置参数
        # config = request.get_json() or {}

        localTime = time.strftime("%Y%m%d%H%M%S")
        """
        安全的任务启动器，处理参数传递
        """
        args = None
        kwargs = None
        task_id = None
        # 安全处理参数
        safe_args = args if args is not None else ()
        safe_kwargs = kwargs if kwargs is not None else {}

        # 初始化任务状态
        with task_lock:
            running_tasks[task_id] = {
                'running': True,
                'start_time': localTime,
                'config': None,
                'execution_count': 0,
                'last_execution': None,
                'thread': None
            }

        # 在后台线程中启动任务
        thread = threading.Thread(
            target=permanent_task,
            args=safe_args,
            kwargs=safe_kwargs,
            daemon=True  # 设置为守护线程，主程序退出时自动结束
        )
        thread.start()

        # 保存线程引用
        with task_lock:
            running_tasks[task_id]['thread'] = thread

        logger.info(f"任务 {task_id} 已启动")

        # 立即返回响应
        return jsonify({
            'status': 'success',
            'message': '任务已启动',
            'task_id': task_id,
            'start_time': running_tasks[task_id]['start_time']
        }), 202  # 202 Accepted 表示请求已接受但处理尚未完成

    except Exception as e:
        logger.error(f"启动任务失败: {e}")
        return jsonify({
            'status': 'error',
            'message': f'启动任务失败: {str(e)}'
        }), 500




# 根据数据库配置加载读取任务开始记录相关数据
def init_task_zb():
    # tj_jg()

    logger.info(f'执行定时任务初始化')
    # 4. 复杂的Cron表达式
    scheduler.add_job(
        tj_jg,
        # trigger=CronTrigger.from_crontab('1,6,11,16,21,26,31,36,41,46,51,56 9-14 * * 1-5'),  # 工作日9点、14点、18点

        # trigger=CronTrigger.from_crontab('1,6,11,16,21,26,31,36,41,46,51,56 9-15 * * 0-6'),  # 工作日9点、14点、18点
        trigger=CronTrigger(minute='*/1'),
        id='task_1'
    )
    scheduler.start()
    logger.info(f'执行定时任务初始化完成')

    try:
        # 保持程序运行
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        scheduler.shutdown()
        logger.info("调度器已关闭")



def tj_jg():
    # 1. 获取数据库配置 根据不同策略规则获取对应的数据  是否写数据库 还是写对象存储
    # 1.设置需要跑的股票代码加入到股票池，然后进行数据回测跑分，特别是 5分 / 30分 大概第6分就要跑一下尽快评估结果进行消息写入并推送
    result = db.stock_follow_query()
    localTime =time.strftime("%Y-%m-%d %H:%M:%S")
    logger.info(f"当前任务执行时间:{localTime}")
    mair = mr.MaiRuiData()
    #获取当前分钟数
    minTime = time.strftime("%M")
    minTimeNumber = int(minTime)
    for row in result:

        # 需要屏蔽 d m y 取余要报错
        time_period =row['time_period']
        time_period_number = int(time_period)

        # 当取余数==1 说明就是 5n+1 执行 5f  15n+1 执行15f  30n+1 执行30f
        ys =minTimeNumber % time_period_number
        print(f"定时任务执行记录。。。。。。")
        if(ys<4 and ys>0 and False):
        # if(True):
            #计算类型 1 接口  2 oss
            analysis_method = row['analysis_method']
            logger.info(f"  用户: {row['stock_code']},计算kdj指标值")
            if analysis_method==1:
                # 可以像字典一样访问，也可以像元组一样访问
                # print(f"  元组访问: {row[0]}, {row[1]}, {row[2]}")  上层数据库不转译 则可以用数组取
                sourceDf = mair.mai_rui_data_k(row['stock_code'], row['time_period'], row['last_quantity'])

            if analysis_method==2:
                sourceDf = oss_soure_data(row)


            strategy_calculation(sourceDf, row)


    # print(result)


def strategy_calculation(df,row):
    txd = tx.TxdClass()
    staraList = row['indicators_list'].split(",")
    res =f"课程编码:{row['stock_code']},触发告警:\n"
    alarm_state= False

    for s in staraList:
        # 看看策略id是否包含kdj编码,如果包含则进行数据计算
        if "kdj" == s:
            # 加载策略id 类型 是不是kdj 如果是 则继续进行计算
            # 通信达kdj计算适配
            kdj_res =txd.txd_kdj(df,9,3,3,row)




        if "macd" == s and alarm_state:
            # 赫尔均线
            close_prices21 = df['c'].values  # 获取收盘价序列
            close_prices88 = df['c'].values  # 获取收盘价序列
            x21 = txd.tdx_double_wma_single_param(close_prices21, 21)
            res = res+x21
            x88 = txd.tdx_double_wma_single_param(close_prices88, 88)
            res = res+x88

        if "rsi"  == s and alarm_state:
            # 加载策略id 类型 是不是kdj 如果是 则继续进行计算
            # 通信达kdj计算适配
            kdj_res =txd.txd_kdj(df,9,3,3,row)

    # if(res):
    #     #告警通知推送
    #     print(f"策略提示结果如下:\n{res}")

    #
    #
    #
    # close_prices88 = df88['c'].values  # 获取收盘价序列
    #
    # x =macd.tdx_double_wma_single_param(close_prices21,21)
    # df21['macd21'] = x
    #
    # y =macd.tdx_double_wma_single_param(close_prices88,88)
    # df88['macd88'] = y


